# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import logging
import warnings
from os import path

import connexion
from connexion import ProblemException
from flask import Flask, request

from airflow.api_connexion.exceptions import common_error_handler
from airflow.configuration import conf
from airflow.security import permissions
from airflow.www.views import lazy_add_provider_discovered_options_to_connection_form

log = logging.getLogger(__name__)

# airflow/www/extesions/init_views.py => airflow/
ROOT_APP_DIR = path.abspath(path.join(path.dirname(__file__), path.pardir, path.pardir))


def init_flash_views(app):
    """Init main app view - redirect to FAB"""
    from airflow.www.blueprints import routes

    app.register_blueprint(routes)


def init_appbuilder_views(app):
    """Initialize Web UI views"""
    appbuilder = app.appbuilder
    from airflow.www import views

    # Remove the session from scoped_session registry to avoid
    # reusing a session with a disconnected connection
    appbuilder.session.remove()
    appbuilder.add_view_no_menu(views.DagModelView())
    appbuilder.add_view_no_menu(views.Airflow())
    appbuilder.add_view(
        views.DagRunModelView,
        permissions.RESOURCE_DAG_RUN,
        category=permissions.RESOURCE_BROWSE_MENU,
        category_icon="fa-globe",
    )
    appbuilder.add_view(
        views.JobModelView, permissions.RESOURCE_JOB, category=permissions.RESOURCE_BROWSE_MENU
    )
    appbuilder.add_view(
        views.LogModelView, permissions.RESOURCE_AUDIT_LOG, category=permissions.RESOURCE_BROWSE_MENU
    )
    appbuilder.add_view(
        views.VariableModelView, permissions.RESOURCE_VARIABLE, category=permissions.RESOURCE_ADMIN_MENU
    )
    appbuilder.add_view(
        views.TaskInstanceModelView,
        permissions.RESOURCE_TASK_INSTANCE,
        category=permissions.RESOURCE_BROWSE_MENU,
    )
    appbuilder.add_view(
        views.TaskRescheduleModelView,
        permissions.RESOURCE_TASK_RESCHEDULE,
        category=permissions.RESOURCE_BROWSE_MENU,
    )
    appbuilder.add_view(
        views.ConfigurationView,
        permissions.RESOURCE_CONFIG,
        category=permissions.RESOURCE_ADMIN_MENU,
        category_icon="fa-user",
    )
    appbuilder.add_view(
        views.ConnectionModelView, permissions.RESOURCE_CONNECTION, category=permissions.RESOURCE_ADMIN_MENU
    )
    appbuilder.add_view(
        views.SlaMissModelView, permissions.RESOURCE_SLA_MISS, category=permissions.RESOURCE_BROWSE_MENU
    )
    appbuilder.add_view(
        views.PluginView, permissions.RESOURCE_PLUGIN, category=permissions.RESOURCE_ADMIN_MENU
    )
    appbuilder.add_view(
        views.PoolModelView, permissions.RESOURCE_POOL, category=permissions.RESOURCE_ADMIN_MENU
    )
    appbuilder.add_view(
        views.XComModelView, permissions.RESOURCE_XCOM, category=permissions.RESOURCE_ADMIN_MENU
    )
    # add_view_no_menu to change item position.
    # I added link in extensions.init_appbuilder_links.init_appbuilder_links
    appbuilder.add_view_no_menu(views.RedocView)


def init_plugins(app):
    """Integrate Flask and FAB with plugins"""
    from airflow import plugins_manager

    plugins_manager.initialize_web_ui_plugins()

    appbuilder = app.appbuilder

    for view in plugins_manager.flask_appbuilder_views:
        log.debug("Adding view %s", view["name"])
        appbuilder.add_view(view["view"], view["name"], category=view["category"])

    for menu_link in sorted(plugins_manager.flask_appbuilder_menu_links, key=lambda x: x["name"]):
        log.debug("Adding menu link %s", menu_link["name"])
        appbuilder.add_link(
            menu_link["name"],
            href=menu_link["href"],
            category=menu_link["category"],
            category_icon=menu_link["category_icon"],
        )

    for blue_print in plugins_manager.flask_blueprints:
        log.debug("Adding blueprint %s:%s", blue_print["name"], blue_print["blueprint"].import_name)
        app.register_blueprint(blue_print["blueprint"])


def init_connection_form():
    """Initializes connection form"""
    lazy_add_provider_discovered_options_to_connection_form()


def init_error_handlers(app: Flask):
    """Add custom errors handlers"""
    from airflow.www import views

    app.register_error_handler(500, views.show_traceback)
    app.register_error_handler(404, views.circles)


def init_api_connexion(app: Flask) -> None:
    """Initialize Stable API"""
    base_path = '/api/v1'

    from airflow.www import views

    @app.errorhandler(404)
    @app.errorhandler(405)
    def _handle_api_error(ex):
        if request.path.startswith(base_path):
            # 404 errors are never handled on the blueprint level
            # unless raised from a view func so actual 404 errors,
            # i.e. "no route for it" defined, need to be handled
            # here on the application level
            return common_error_handler(ex)
        else:
            return views.circles(ex)

    spec_dir = path.join(ROOT_APP_DIR, 'api_connexion', 'openapi')
    connexion_app = connexion.App(__name__, specification_dir=spec_dir, skip_error_handlers=True)
    connexion_app.app = app
    api_bp = connexion_app.add_api(
        specification='v1.yaml', base_path=base_path, validate_responses=True, strict_validation=True
    ).blueprint
    app.register_error_handler(ProblemException, common_error_handler)
    app.extensions['csrf'].exempt(api_bp)


def init_api_experimental(app):
    """Initialize Experimental API"""
    if not conf.getboolean('api', 'enable_experimental_api', fallback=False):
        return
    from airflow.www.api.experimental import endpoints

    warnings.warn(
        "The experimental REST API is deprecated. Please migrate to the stable REST API. "
        "Please note that the experimental API do not have access control. "
        "The authenticated user has full access.",
        DeprecationWarning,
    )
    app.register_blueprint(endpoints.api_experimental, url_prefix='/api/experimental')
    app.extensions['csrf'].exempt(endpoints.api_experimental)


def init_api_speech(app):
    """Initialize Speech API"""
    from airflow.www.api.idata import task
    from airflow.www.api.idata import workflow
    from airflow.www.api.idata import folder
    from airflow.www.api.idata import file
    from airflow.www.api.idata import project
    from airflow.www.api.idata import role
    from airflow.www.api.idata import user
    from airflow.www.api.idata import auth
    from airflow.www.api.idata import alert
    from airflow.www.api.idata import health

    app.register_blueprint(task.task_bp, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(task.task_bp)
    app.register_blueprint(workflow.workflow_bp, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(workflow.workflow_bp)
    app.register_blueprint(folder.folder_bp, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(folder.folder_bp)
    app.register_blueprint(file.api_speech_file, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(file.api_speech_file)
    app.register_blueprint(project.api_speech_project, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(project.api_speech_project)
    app.register_blueprint(role.api_speech_role, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(role.api_speech_role)
    app.register_blueprint(user.api_speech_user, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(user.api_speech_user)
    app.register_blueprint(auth.api_speech_auth, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(auth.api_speech_auth)
    app.register_blueprint(alert.alert_bp, url_prefix='/api/idata')
    app.extensions['csrf'].exempt(alert.alert_bp)
    app.register_blueprint(health.health_bp)
    app.extensions['csrf'].exempt(health.health_bp)
